/*
 * Copyright (C) 2017 Honeywell, Inc. All Rights Reserved.
 */

package org.thanos.demo.nats;

import io.nats.client.Connection;
import io.nats.client.Dispatcher;
import io.nats.client.MessageHandler;
import io.nats.client.Nats;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.concurrent.TimeoutException;

public class NatsClient {

    private final Connection connection;

    public NatsClient(String natsConnectUrl) throws IOException, InterruptedException {
        this.connection = Nats.connect(natsConnectUrl);
    }

    public void close() throws InterruptedException, TimeoutException {
        this.connection.flush(Duration.ZERO);
        this.connection.close();
    }

    public void publish(String topic, byte[] data) {
        this.connection.publish(topic, data);
    }

    public void publish(String topic, String data) {
        this.connection.publish(topic, data.getBytes(StandardCharsets.UTF_8));
    }

    public void subscribe(String topic, MessageHandler messageHandler) {
        Dispatcher dispatcher = this.connection.createDispatcher(messageHandler);
        dispatcher.subscribe(topic);
    }

    public void subscribe(String topic, String queue, MessageHandler messageHandler) {
        Dispatcher dispatcher = this.connection.createDispatcher(messageHandler);
        dispatcher.subscribe(topic, queue);
    }
}
